
#include "bocat/common/logging.hpp"

#include "boque/tranzmq/channel.hpp"

namespace boque		{
namespace tranzmq	{


///////////////////////////////////////////////////////////////////////
//

Channel::Channel( const std::string& address, const std::string& buf_dir, zmq::context_t* context_p )
	:	_address( address ),
		_buf_dir( buf_dir ),
		_buffesq( _buf_dir ),
		_socket_p( NULL ),
		_status( UNKNOWN ),
		_waiter_p( NULL ),
		_context_p( context_p )
{
	_socket_p = new zmq::socket_t( *_context_p, ZMQ_REQ );
	_socket_connected = false;

	if ( !_buffesq.empty() ) {
		// if queue is not empty, channel is buffered
		_status = BUFFERED;

		bocat::common::Logger*	logger_p = bocat::common::getLogger();
		logger_p->error(__LINE__, __FILE__, "channel to ZeroMQ %s enter BUFFER mode", _address.c_str());
	}

	_relay_enable = false;
}

Channel::~Channel()
{
	if ( NULL != _waiter_p ) {
		delete _waiter_p;
		_waiter_p = NULL;
	}

	if ( NULL != _socket_p ) {
		delete _socket_p;
		_socket_p = NULL;
	}
}


///////////////////////////////////////////////////////////////////////
//

int
Channel::send( const std::string& data )
{
	return	send( data.c_str(), data.size() );
}

int
Channel::send( const char* data, size_t size )
{
	boost::mutex::scoped_lock	_scoped_lock( _mutex );

	if ( UNKNOWN == _status ) {
		// try to connect 
		try {
			_socket_p->connect( _address.c_str() );
			_status = ALIVE;
			_socket_connected = true;
		}
		catch (std::exception& e) {
			bocat::common::Logger*	logger_p = bocat::common::getLogger();
			logger_p->error(__LINE__, __FILE__, "fail to connect ZeroMQ %s -- %s", _address.c_str(), e.what());
			logger_p->error(__LINE__, __FILE__, "channel to ZeroMQ %s enter BUFFER mode", _address.c_str());

			_status = BUFFERED;
			_socket_connected = false;
		}
	}

	if ( ALIVE == _status ) {
		int	rc = 0;
		// try to send data directly
		try {
			zmq::message_t	msg( size );
			memcpy( msg.data(), data, size );
			rc = _socket_p->send( msg );
			if ( rc < 0 ) {
				bocat::common::Logger*	logger_p = bocat::common::getLogger();
				logger_p->error(__LINE__, __FILE__, "fail to send data to ZeroMQ %s ", _address.c_str());
			}

			zmq::message_t	reply;
			rc = _socket_p->recv( &reply );
			if ( rc < 0 ) {
				bocat::common::Logger*	logger_p = bocat::common::getLogger();
				logger_p->error(__LINE__, __FILE__, "fail to recv data from ZeroMQ %s ", _address.c_str());
			}
		}
		catch( std::exception& e ) {
			bocat::common::Logger*	logger_p = bocat::common::getLogger();
			logger_p->error(__LINE__, __FILE__, "fail to send data to ZeroMQ %s -- %s", _address.c_str(), e.what());
			logger_p->error(__LINE__, __FILE__, "channel to ZeroMQ %s enter BUFFER mode", _address.c_str());

			_status = BUFFERED;
			_socket_connected = false;
		}
	}

	if ( BUFFERED == _status ) {
		// push data to disk queue
		_buffesq.push( data, size );
	}

	return	0;
}

///////////////////////////////////////////////////////////////////////
//

int
Channel::relay( void )
{
	if ( !_relay_enable ) {
		return	0;
	}

	boost::mutex::scoped_lock	_scoped_lock( _mutex_relay );

	if ( ALIVE == _status ) {
		return	0;
	}

	if ( false == _socket_connected ) {
		// try to connect 
		try {
			_socket_p->connect( _address.c_str() );
			_socket_connected = true;
		}
		catch (std::exception& e) {
			//bocat::common::Logger*	logger_p = bocat::common::getLogger();
			//logger_p->error(__LINE__, __FILE__, "fail to connect ZeroMQ %s -- %s", _address.c_str(), e.what());
			//logger_p->error(__LINE__, __FILE__, "channel to ZeroMQ %s enter BUFFER mode", _address.c_str());

			_status = BUFFERED;
			_socket_connected = false;
		}
	}
	if ( false == _socket_connected ) {
		return	0;
	}

	// send data in queue
	int	rc = do_relay();
	if ( rc < 0 ) {
		return	-1;
	}

	// send done
	// try to set status as alive
	{
		boost::mutex::scoped_lock	_scoped_lock( _mutex );

		// TODO : to clear queue
		int	rc = do_relay();
		if ( rc < 0 ) {
			return	-1;
		}

		if ( ALIVE != _status ) {
			_status = ALIVE;

			bocat::common::Logger*	logger_p = bocat::common::getLogger();
			logger_p->error(__LINE__, __FILE__, "channel to ZeroMQ %s enter ALIVE mode", _address.c_str());
		}
	}

	return	0;
}


int
Channel::do_relay( void )
{
	while ( !_buffesq.empty() ) {
		if ( !_relay_enable ) {
			return	-2;
		}

		std::string	data = _buffesq.front();

		// try to send data directly
		try {
			zmq::message_t	msg( data.size() );
			memcpy( msg.data(), data.c_str(), data.size() );
			_socket_p->send( msg );

			zmq::message_t	reply;
			_socket_p->recv( &reply );
		}
		catch( std::exception& e ) {
			//bocat::common::Logger*	logger_p = bocat::common::getLogger();
			//logger_p->error(__LINE__, __FILE__, "fail to send data to ZeroMQ %s -- %s", _address.c_str(), e.what());
			//logger_p->error(__LINE__, __FILE__, "channel to ZeroMQ %s enter BUFFER mode", _address.c_str());

			_status = BUFFERED;
			_socket_connected = false;

			return	-1;
		}

		// just discard pop result
		data = _buffesq.pop();
	}

	return	0;
}


///////////////////////////////////////////////////////////////////////
//

}	//// namespace tranzmq
}	//// namespace boque

